package cn.bigdata.sparkcore.job

import org.apache.spark.{SparkConf, SparkContext}

object Case3 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName(this.getClass.getName.init)
    val sc = new SparkContext(conf)
    sc.setLogLevel("warn")

    val clickLog =
      """INFO 2019-09-01 00:29:53 requestURI:/click?app=1&p=1&adid=18005472&industry=469&adid=31
        |INFO 2019-09-01 00:30:31 requestURI:/click?app=2&p=1&adid=18005472&industry=469&adid=31
        |INFO 2019-09-01 00:31:03 requestURI:/click?app=1&p=1&adid=18005472&industry=469&adid=32
        |INFO 2019-09-01 00:31:51 requestURI:/click?app=1&p=1&adid=18005472&industry=469&adid=33""".stripMargin

    val impLog =
      """INFO 2019-09-01 00:29:53 requestURI:/imp?app=1&p=1&adid=18005472&industry=469&adid=31
        |INFO 2019-09-01 00:29:53 requestURI:/imp?app=1&p=1&adid=18005472&industry=469&adid=31
        |INFO 2019-09-01 00:29:53 requestURI:/imp?app=1&p=1&adid=18005472&industry=469&adid=34""".stripMargin

    val clickLogRdd = sc.makeRDD(clickLog.split("\n")).map(line => {
      val arr = line.trim.split("&")
      (arr.last, 1)
    }).reduceByKey(_ + _).coalesce(1)

    val impLogRdd = sc.makeRDD(impLog.split("\n")).map(line => {
      val arr = line.trim.split("&")
      (arr.last, 1)
    }).reduceByKey(_ + _).coalesce(1)
    val impLogBC = sc.broadcast(impLogRdd.collect())

    sc.makeRDD(impLogBC.value).fullOuterJoin(clickLogRdd).mapValues {
      case (imp, click) => (imp.getOrElse(0), click.getOrElse(0))
    }.map(v => s"${v._1},${v._2._1},${v._2._2}")
      .saveAsTextFile("hdfs://centos1:8020/implog")

    //
    //    val impLogBC = sc.broadcast(impLogRdd.collectAsMap())
    //    val clickLogBC = sc.broadcast(clickLogRdd.collectAsMap())
    //
    //    val keys = impLogBC.value.keySet ++ clickLogBC.value.keySet
    //    var result = ""
    //    keys.foreach(k =>{
    //      result = result + s"$k,${impLogBC.value.getOrElse(k, 0)},${clickLogBC.value.getOrElse(k, 0)}\n"
    //    })
    //    println(result)
        Thread.sleep(1000000)


    sc.stop()
  }
}
